package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Spark_stream_Window2 {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("window")
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")
    ssc.checkpoint("BD1")
    val lines = ssc.socketTextStream("localhost",9999)
    val wordOne = lines.map( (_,1))

    /*
        reduceBykeAndWindow:当窗口范围比较大，滑动幅度比较小，可以采用增加数据和删除数据的方法
        没有了重复数据，也不会进行重复计算
     */
    val winDS = wordOne.reduceByKeyAndWindow(
      (x: Int, y: Int) => {
        x + y
      },
      (x: Int, y: Int) => {
        x - y
      },
      Seconds(9), Seconds(3)
    )

  //Spark Streaming 没有输出的情况，在运行时会报错：No output operations registered, so nothing to execute
    //winDS.print()// 在输出的时候会打印时间戳  DStream . print  带有时间戳
    winDS.foreachRDD(rdd=>{//DStream类型 转换为 rdd
      rdd.collect().foreach(println)///Array.println
    })

   ssc.start()
   ssc.awaitTermination()
  }

}
